#include "../test_helpers.h"
#include "MessageConsumer.h"
#include <spdlog/spdlog.h>
#include "MessageProducer.h"
#include <gtest/gtest.h>
#include <atomic>

using namespace MessageFramework;
using namespace MessageFramework::Test;

class MessageConsumerTest : public RedisTestBase {
protected:
    std::string secretKey = "test-secret-key-32bytes-long!!!";
    std::string streamName;
    std::string consumerGroup;

    void SetUp() override {
        RedisTestBase::SetUp();
        streamName = testStreamPrefix + "_consumer";
        consumerGroup = testStreamPrefix + "_group";
    }
};

// ==================== 基础功能测试 ====================

TEST_F(MessageConsumerTest, StartStop) {
    MessageConsumer::Config config;
    config.streamName = streamName;
    config.consumerGroup = consumerGroup;
    config.consumerId = "TestConsumer";
    config.autoCreateGroup = true;

    MessageConsumer consumer(redisPool, config);

    EXPECT_FALSE(consumer.isRunning());

    consumer.start();
    EXPECT_TRUE(consumer.isRunning());

    consumer.stop();
    EXPECT_FALSE(consumer.isRunning());
}

TEST_F(MessageConsumerTest, ConsumeMessage) {
    // 创建生产者发送消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";
    producerConfig.secretKey = secretKey;

    MessageProducer producer(redisPool, producerConfig);

    // 创建消费者
    MessageConsumer::Config consumerConfig;
    consumerConfig.streamName = streamName;
    consumerConfig.consumerGroup = consumerGroup;
    consumerConfig.consumerId = "TestConsumer";

    auto dispatcher = std::make_shared<MessageDispatcher>();
    MessageDispatcher::Config dispatcherConfig;
    dispatcherConfig.secretKey = secretKey;
    dispatcher->setConfig(dispatcherConfig);

    auto handler = std::make_shared<MockHandler>("TestMessage");
    dispatcher->registerHandler("TestMessage", handler);

    MessageConsumer consumer(redisPool, consumerConfig, dispatcher);
    consumer.start();

    // 发送消息
    producer.send(streamName, "TestMessage", "{\"test\":\"data\"}");

    // 等待消息被消费
    bool received = TestHelpers::waitFor([&]() {
        return handler->getCallCount() > 0;
    }, 5000);

    EXPECT_TRUE(received);
    EXPECT_EQ(handler->getCallCount(), 1);

    consumer.stop();
}

// ==================== 批量消费测试 ====================

TEST_F(MessageConsumerTest, BatchConsume) {
    // 预先发送多条消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";

    MessageProducer producer(redisPool, producerConfig);

    const int messageCount = 20;
    for (int i = 0; i < messageCount; ++i) {
        producer.send(streamName, "TestMessage", "{\"index\":" + std::to_string(i) + "}");
    }

    // 创建消费者（批量读取）
    MessageConsumer::Config consumerConfig;
    consumerConfig.streamName = streamName;
    consumerConfig.consumerGroup = consumerGroup;
    consumerConfig.consumerId = "TestConsumer";
    consumerConfig.batchSize = 10;  // 每次读取10条

    auto dispatcher = std::make_shared<MessageDispatcher>();
    auto handler = std::make_shared<MockHandler>("TestMessage");
    dispatcher->registerHandler("TestMessage", handler);

    MessageConsumer consumer(redisPool, consumerConfig, dispatcher);
    consumer.start();

    // 等待所有消息被消费
    bool received = TestHelpers::waitFor([&]() {
        return handler->getCallCount() >= messageCount;
    }, 10000);

    EXPECT_TRUE(received);
    EXPECT_EQ(handler->getCallCount(), messageCount);

    consumer.stop();
}

// ==================== 消息过滤测试 ====================

TEST_F(MessageConsumerTest, FilterByConsumerId) {
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";

    MessageProducer producer(redisPool, producerConfig);

    // 创建消费者（启用过滤）
    MessageConsumer::Config consumerConfig;
    consumerConfig.streamName = streamName;
    consumerConfig.consumerGroup = consumerGroup;
    consumerConfig.consumerId = "Consumer1";
    consumerConfig.filterByConsumerId = true;

    auto dispatcher = std::make_shared<MessageDispatcher>();
    auto handler = std::make_shared<MockHandler>("TestMessage");
    dispatcher->registerHandler("TestMessage", handler);

    MessageConsumer consumer(redisPool, consumerConfig, dispatcher);
    consumer.start();

    // 发送定向消息（发给Consumer1）
    producer.send(streamName, "TestMessage", "{\"test\":\"1\"}", "Consumer1");

    // 发送定向消息（发给Consumer2）
    producer.send(streamName, "TestMessage", "{\"test\":\"2\"}", "Consumer2");

    // 发送广播消息（consumerId为空）
    producer.send(streamName, "TestMessage", "{\"test\":\"3\"}", "");

    // 等待消息被处理
    std::this_thread::sleep_for(std::chrono::seconds(2));

    // Consumer1应该收到2条消息：1条定向消息 + 1条广播消息
    EXPECT_EQ(handler->getCallCount(), 2);

    auto stats = consumer.getStats();
    EXPECT_EQ(stats.messagesFiltered, 1);  // Consumer2的消息被过滤

    consumer.stop();
}

// ==================== 统计信息测试 ====================

TEST_F(MessageConsumerTest, Statistics) {
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";

    MessageProducer producer(redisPool, producerConfig);

    MessageConsumer::Config consumerConfig;
    consumerConfig.streamName = streamName;
    consumerConfig.consumerGroup = consumerGroup;
    consumerConfig.consumerId = "TestConsumer";

    auto dispatcher = std::make_shared<MessageDispatcher>();
    auto handler = std::make_shared<MockHandler>("TestMessage");
    dispatcher->registerHandler("TestMessage", handler);

    MessageConsumer consumer(redisPool, consumerConfig, dispatcher);
    consumer.start();

    // 发送5条消息
    for (int i = 0; i < 5; ++i) {
        producer.send(streamName, "TestMessage", "{\"index\":" + std::to_string(i) + "}");
    }

    // 等待消息被处理
    TestHelpers::waitFor([&]() {
        return handler->getCallCount() >= 5;
    }, 5000);

    // 检查统计信息
    auto stats = consumer.getStats();
    EXPECT_EQ(stats.messagesReceived, 5);
    EXPECT_EQ(stats.messagesProcessed, 5);
    EXPECT_EQ(stats.messagesFailed, 0);

    consumer.stop();
}

// ==================== Handler注册测试 ====================

TEST_F(MessageConsumerTest, RegisterHandler) {
    MessageConsumer::Config config;
    config.streamName = streamName;
    config.consumerGroup = consumerGroup;
    config.consumerId = "TestConsumer";

    MessageConsumer consumer(redisPool, config);

    auto handler = std::make_shared<MockHandler>("TestMessage");
    consumer.registerHandler(handler);

    // 启动消费者
    consumer.start();

    // 发送消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";
    MessageProducer producer(redisPool, producerConfig);

    producer.send(streamName, "TestMessage", "{\"test\":\"data\"}");

    // 等待消息被处理
    bool received = TestHelpers::waitFor([&]() {
        return handler->getCallCount() > 0;
    }, 5000);

    EXPECT_TRUE(received);

    consumer.stop();
}

// ==================== 回调函数测试 ====================

TEST_F(MessageConsumerTest, MessageCallback) {
    MessageConsumer::Config config;
    config.streamName = streamName;
    config.consumerGroup = consumerGroup;
    config.consumerId = "TestConsumer";

    MessageConsumer consumer(redisPool, config);

    std::atomic<int> callbackCount(0);
    consumer.setMessageCallback([&](const Message& msg) -> int {
        callbackCount++;
        return 0;
    });

    consumer.start();

    // 发送消息
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "TestProducer";
    MessageProducer producer(redisPool, producerConfig);

    producer.send(streamName, "TestMessage", "{\"test\":\"data\"}");

    // 等待回调被调用
    bool called = TestHelpers::waitFor([&]() {
        return callbackCount.load() > 0;
    }, 5000);

    EXPECT_TRUE(called);
    EXPECT_EQ(callbackCount.load(), 1);

    consumer.stop();
}

int main(int argc, char** argv) {
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
